package juc;

import org.junit.platform.commons.util.StringUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ProdConsumer_BlockQueueDemo {
	public static void main(String args[]) throws InterruptedException {
		MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
		new Thread(() -> {
			System.out.println("启动生产。。。");
			try {
				myResource.product();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}, "PRODUCT").start();
		new Thread(() -> {
			System.out.println("启动消费。。。");
			try {
				myResource.consume();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}, "CONSUMER").start();
		TimeUnit.SECONDS.sleep(20);
		myResource.stop();
	}

}

class MyResource {
	private volatile boolean FLAG = true;
	private AtomicInteger atomicInteger = new AtomicInteger();
	private BlockingQueue<String> blockingQueue;

	public MyResource(BlockingQueue<String> blockingQueue) {
		this.blockingQueue = blockingQueue;
		System.out.println(blockingQueue.getClass().getName());
	}

	public void product() throws InterruptedException {
		while (FLAG) {
			if (blockingQueue.offer(atomicInteger.incrementAndGet() + "", 2, TimeUnit.SECONDS)) {
				System.out.println("生产成功: " + atomicInteger.get() + "; 当前共" + blockingQueue.size());
			} else {
				System.out.println("生产失败: " + atomicInteger.get());
			}
			TimeUnit.SECONDS.sleep(1);
		}
		System.out.println("===========退出===========");
	}

	public void consume() throws InterruptedException {
		while (FLAG) {
			if (StringUtils.isBlank(blockingQueue.poll(2, TimeUnit.SECONDS))) {
				System.out.println("消费失败: " + atomicInteger.get() + "; 当前共" + blockingQueue.size());
				return;
			}
			System.out.println("消费成功: " + atomicInteger.get());
		}
	}

	public void stop() {
		FLAG = false;
	}
}